package cn.doitedu.demo6;

import cn.doitedu.beans.UserAction;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.io.IOException;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/10
 * @Desc: 学大数据，上多易教育
 * <p>
 * {
 * "rule_id":"rule-m05-001",
 * "past_future_condition":{
 * "event_id":"K",
 * "start_time":1710054281000,
 * "end_time":1710054281000,
 * "min_count":3,
 * "condition_id":"pfc01"
 * },
 * "trigger_event_id":"M"
 * }
 **/
public class RuleModelCalculator_05 implements RuleCalculator {

    private String ruleId;
    Jedis jedis;

    long preStatisticSplitPoint;

    ValueState<Integer> countState;

    String triggerEventId;

    String targetEventId;
    long endTime;
    int minCount;
    String conditionId;

    JSONObject message;

    @Override
    public String ruleId() {
        return this.ruleId;
    }

    @Override
    public void init(RuntimeContext runtimeContext, RuleMeta ruleMeta) throws IOException {

        String ruleParamJson = ruleMeta.getRule_param_json();
        JSONObject jsonObject = JSON.parseObject(ruleParamJson);
        this.ruleId = jsonObject.getString("rule_id");

        this.triggerEventId = jsonObject.getString("trigger_event_id");

        JSONObject pastFutureCondition = jsonObject.getJSONObject("past_future_condition");
        this.targetEventId = pastFutureCondition.getString("event_id");
        this.endTime = pastFutureCondition.getLongValue("end_time");
        this.minCount = pastFutureCondition.getIntValue("min_count");
        this.conditionId = pastFutureCondition.getString("condition_id");

        this.countState = runtimeContext.getState(new ValueStateDescriptor<Integer>(ruleId + "-cnt", Integer.class));
        this.jedis = new Jedis("doitedu", 6379);

        // 取出历史分界点
        this.preStatisticSplitPoint = ruleMeta.getPre_statistic_split_point();

        this.message = new JSONObject();
        message.put("rule_id", ruleId);

    }

    @Override
    public void calc(UserAction currentAction, Collector<String> out) throws IOException {

        if (currentAction.getAction_time() <= preStatisticSplitPoint
                || currentAction.getAction_time() > endTime
                || (!currentAction.getEvent_id().equals(targetEventId) && !currentAction.getEvent_id().equals(triggerEventId)))
            return;


        if (currentAction.getEvent_id().equals(targetEventId)) {
            // 能走到这里，说明当前currentAction是满足时间条件,则进行计算
            Integer stateCountValue = countState.value();

            // 如果计数状态中的值为null，说明当前行为的行为人还没有被统计过，则需要去redis中查询他的历史段初始值
            if (stateCountValue == null) {
                // redis中历史值：hash结构
                //     大key  ==>   rule_id
                //     value  ==>       小key => 条件id:用户id   值: 5
                String redisValueStr = jedis.hget(ruleId, conditionId + ":" + currentAction.getUser_id());
                int redisValue = redisValueStr == null ? 0 : Integer.parseInt(redisValueStr);

                // 将redis查到的值+1，更新到计数状态
                countState.update(redisValue + 1);

            } else {
                countState.update(stateCountValue + 1);

            }

        } else {
            // 代码走到这里，一定是触发行为
            int count = 0;

            // 先获取该用户的统计事件次数
            if (countState.value() == null) {
                // 如果状态为空，得去redis查历史值
                String redisValueStr = jedis.hget(ruleId, conditionId + ":" + currentAction.getUser_id());
                count = redisValueStr == null ? 0 : Integer.parseInt(redisValueStr);

                // 将查询的值，放入状态
                countState.update(count);

            } else {
                // 如果状态不为空，直接就是状态中的值
                count = countState.value();
            }

            if (count > minCount) {

                message.put("user_id", currentAction.getUser_id());
                message.put("hit_time", currentAction.getAction_time());
                out.collect(message.toJSONString());

            }

        }


    }
}
